#include <config.h>
#include <core.h>

extern  msgd_event_t  *msgd_posted_events;
//extern  block_queue_t  msgd_posted_events_queue;

static  msgd_int_t  msgd_epoll_add_event(msgd_event_t *ev, msgd_int_t event, msgd_int_t flags);
static  msgd_int_t  msgd_epoll_del_event(msgd_event_t *ev, msgd_int_t event, msgd_int_t flags);

#if 0
static  msgd_int_t  msgd_epoll_process_changes_event(msgd_uint_t nowait);
#endif
static  msgd_int_t  msgd_epoll_process_events(msgd_int_t timer, msgd_int_t flags);

static  msgd_int_t  msgd_epoll_init();
static  void        msgd_epoll_done();


msgd_event_actions_t msgd_event_actions = {
        msgd_epoll_add_event,
        msgd_epoll_del_event,
        NULL,//msgd_epoll_process_changes_event, 
    	msgd_epoll_process_events,
    	msgd_epoll_init,
    	msgd_epoll_done
};

static int     ep           = -1;
static struct  epoll_event  *event_list;
static msgd_uint_t  nevents = 1024;




static  msgd_int_t  msgd_epoll_add_event(msgd_event_t *ev, msgd_int_t event, msgd_int_t flags)
{
   int                 op;
   msgd_int_t          events , prev;
   msgd_connection_t  *c;
   msgd_event_t       *e;
   struct epoll_event ee;
   
   DEBUG("msgd_epoll_add_event ev:%p event:%d flags:%d",ev,event,flags);

   c = ev->data;
   events = event;
  
   if(event == MSGD_READ_EVENT){
     e = c->write; 
     prev = EPOLLOUT;
   }else{
     e = c->read;
     prev = EPOLLIN;
   }

   if(e->active){
     op = EPOLL_CTL_MOD;
     events |= prev; 
   }else{
     op = EPOLL_CTL_ADD;
   }

   ee.events   = events | flags;
   ee.data.ptr = (void*)c;

   DEBUG("epoll add event: fd:%d op:%d ev:%08X",c->fd,op,ee.events);

   if(epoll_ctl(ep,op,c->fd,&ee) == -1){
     FATAL("epoll_ctl(%d,%d) failed",c->fd,op);
     return MSGD_ERROR;
   }
 
   ev->active = 1;
   return MSGD_OK;
}

static  msgd_int_t  msgd_epoll_del_event(msgd_event_t *ev, msgd_int_t event, msgd_int_t flags)
{
   int                 op;
   msgd_int_t          prev;
   msgd_connection_t  *c;
   msgd_event_t       *e;
   struct epoll_event  ee;
 
   DEBUG("msgd_epoll_del_event ev:%p event:%d flags:%d",ev,event,flags);

   if(flags & MSGD_CLOSE_EVENT){
      ev->active = 0; 
      return MSGD_OK;
   }

   c = ev->data;
   
   if(event == MSGD_READ_EVENT){
     e = c->write;
     prev = EPOLLOUT;
   }else{
     e = c->read;
     prev = EPOLLIN;
   }

   if(e->active){
     op = EPOLL_CTL_MOD;
     ee.events = prev | flags;
     ee.data.ptr = (void*)c;
   }else{
     op = EPOLL_CTL_DEL;
     ee.events = 0;
     ee.data.ptr = NULL;
   }
   
   if(epoll_ctl(ep,op,c->fd,&ee) == -1){ 
     ERROR("epoll_ctl(%d,%d) failed",c->fd,op); 
     return MSGD_ERROR;
   }

   ev->active = 0;
   return MSGD_OK;
}


#if 0
static  msgd_int_t  msgd_epoll_process_changes_event(msgd_uint_t nowait)
{
   DEBUG("msgd_epoll_process_changes_event nowait:%d",nowait);
   msgd_int_t n;
   return n;
}
#endif

static  msgd_int_t  msgd_epoll_process_events(msgd_int_t timer, msgd_int_t flags)
{
   int             events , err;
   uint32_t        revents;
   msgd_int_t      i;
   msgd_event_t    *rev, *wev;
   msgd_connection_t *c;

   DEBUG("epoll timer:%d flags:%d",timer,flags);

   events =  epoll_wait(ep,event_list,nevents,timer);
   err    =  (events == -1) ? errno : 0;

   if(flags & MSGD_PROCESS_EVENT_UPDATE_TIME){
      msgd_time_update(); 
   }

   if(err){
     if(err == EINTR){
       WARN("epoll_wait() interrupted");
       return MSGD_OK;
     }
     ERROR("epoll_wait() failed");
     return MSGD_ERROR;
   }

   if(events == 0){
      if(timer != -1){
         DEBUG("epoll_wait() normal timeout"); 
         return MSGD_OK;
      }
      ERROR("epoll_wait() returned no events without timeout");
      return MSGD_ERROR;
   }

   for(i=0; i<events; i++){
     c  = event_list[i].data.ptr;

     if(c->fd == -1){
       DEBUG("epoll: stale event %p",c);
       continue;
     }


     revents = event_list[i].events; 

     DEBUG("epoll: fd:%d ev:%04x d:%p",c->fd,revents,event_list[i].data.ptr);

     if(revents & (EPOLLERR|EPOLLHUP) ){
       ERROR("epoll_wait error fd:%d ev:%04x ",c->fd,revents);
       revents |= EPOLLIN|EPOLLOUT;
       goto WWW;
     }

     rev = c->read;

     if((revents & MSGD_READ_EVENT) && rev->active ){
       if((flags & MSGD_PROCESS_EVENT_POST_EVENTS)){
          if(global.worker_threads_n > 0){
             block_queue_push(&msgd_posted_events_queue,&rev->queue);
          }else{
             msgd_locked_post_event(rev,&msgd_posted_events); 
          }
       }else{
          rev->handler(rev); 
       }
     }

     WWW:
     wev = c->write;

     if((revents & MSGD_WRITE_EVENT) && wev->active ){
       if((flags & MSGD_PROCESS_EVENT_POST_EVENTS)){
          if(global.worker_threads_n > 0){
             block_queue_push(&msgd_posted_events_queue,&wev->queue);
          }else{
             msgd_locked_post_event(wev,&msgd_posted_events); 
          }
       }else{
        wev->handler(wev); 
       }
     }
   }
   return MSGD_OK;
}

static msgd_int_t  msgd_epoll_init()
{
   DEBUG("msgd_epoll_init");
   if(ep == -1){
     ep = epoll_create(nevents / 2); //half all connection 
     if(ep == -1){
        ERROR("epoll_create failed");
        return MSGD_ERROR;
     }
   }

   if(event_list == NULL){
      if(nevents == 0) nevents = 1024;

      event_list = malloc( sizeof(struct epoll_event) * nevents );
      if(event_list == NULL){
        ERROR("msgd_epoll_init malloc failed not enough memory");
        return MSGD_ERROR;
      } 
   }
   
   if(msgd_event_timer_init() == MSGD_ERROR){
      ERROR("timer init failed");
      return MSGD_ERROR;
   }
   return MSGD_OK;
}

static  void        msgd_epoll_done()
{
   DEBUG("msgd_epoll_done");
   if(close(ep) == -1){
      ERROR("epoll close() failed");
   }

   ep = -1;
  
   if(event_list){
      free(event_list);
      event_list = NULL;
   }

   nevents = 0;
}

#if 0
int epoll_read(int fd,char **buf,int *len)
{
   //char *buf = *_buf; 
   printf("inner  methond buf:%p\n",*buf);
   int readn=0, total=0, avail = *len, oldsize= *len;
   *buf = (char*)thread_calloc(avail); 
   if(!*buf){
     WARN("thread_calloc error");
     return -2;
   }
   printf("inner  methond2 buf:%p\n",*buf);
   while(1){
      if(readn == avail){
        *buf = thread_realloc(*buf,oldsize,oldsize+avail);
        if(!*buf){
          WARN("thread_realloc error");
          return -2;
        }
        printf("inner  methond3 buf:%p\n",*buf);
        oldsize += avail;
      }
      readn = read(fd,*buf+total,avail); 
      if(readn<0){
        if(errno == EAGAIN || errno == EWOULDBLOCK){
           WARN("read(%d) error again",fd);
           break;
        }
        INFO("read(%d) error",fd);
        return -1;
      }else if(readn == 0){
        printf("has read zero total:%d\n",total);
        close(fd);
        return -1;
      }else if(readn > 0){
        total += readn; 
        if(readn == avail)
           continue;
        else
           break;
        printf("has read %d\n",readn);
      }
   }
   printf("has read at end total:%d\n",total);
   *len = total;
   return total; 
}

int epoll_wait_new_connect(int ep,int listen_fd,struct epoll_event *events,int maxevents,int timeout){
   int fdcnt = epoll_wait(ep,events,maxevents,timeout);
   int i , clientfd;
   
   for(i=0;i<fdcnt;i++){
      if(events[i].data.fd == listen_fd) {
         struct sockaddr_in client_addr;
         socklen_t  socklen;
         clientfd = accept(listen_fd,(struct sockaddr*)&client_addr,&socklen);
         if(clientfd < 0){
            WARN("accept new connect err:");
            continue;
         } 

         setnonblocking(clientfd);
         
         epoll_add(ep,clientfd,EPOLLIN,EPOLLET);
      }/*else if(events[i].events & EPOLLIN){
         if((sockfd = events[i].data.fd)<0){
           err_msg("invalid new fd");
           continue; 
         }
       
         if
         
      }else if(events[i].events & EPOLLOUT){
      }*/ 
   } 
   return fdcnt;
}
#endif
